package cn.edu.cuit.elena.db;

import java.sql.DatabaseMetaData;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.Map;

public class Schema
{
    private Map<String, Table> tables;
    private DatabaseMetaData metaData;
    private DataSource dataSource;
    private String name;
    private String userContextName;
    private int TableAcount;

    public enum DATA_TYPE
    {
        BIT, INT, SMALLINT, TINYINT, NUMERIC, DECIMAL, MONEY, SMALLMONEY, FLOAT, REAL, DATETIME, SMALLDATETIME, CURSOR, TIMESTAMP, UNIQUEIDENTIFIER, CHAR, VARCHAR, TEXT, NCHAR, NVARCHAR, NTEXT, BINARY, VARBINARY, IMAGE
        ;
    }

    public Schema(DataSource dataSource) throws SQLException
    {
        this.dataSource = dataSource;
        tables = new HashMap<String, Schema.Table>();
        metaData();
        construct();
    }

    public Schema(DataSource dataSource, String userContextName) throws SQLException
    {
        this.dataSource = dataSource;
        this.userContextName = userContextName;
        tables = new HashMap<String, Schema.Table>();
        metaData();
        construct();
    }

    private void construct() throws SQLException
    {
        name = dataSource.getName();

        //get all tables in one schema.
        ResultSet resultSet = metaData.getTables( null, "%", "%", new String[] { "TABLE" } );
        while( resultSet.next() )
        {

            TableAcount++;
            String tableName = resultSet.getString( "TABLE_NAME" );

            //create new table.
            Table table = new Table( tableName );
            if( userContextName != null )
            {
                table.setHiveTableName( userContextName + "_" + dataSource.getName() + "_" + tableName );
                table.setHdfsPath( "/user/quintus/" + userContextName + "/" + dataSource.getName() + "/" + tableName );
            }

            //Row count.
            ResultSet resultSet0 = dataSource.getConnection().createStatement()
                .executeQuery( "select count(*) from " + tableName );
            resultSet0.next();
            long rowCount = resultSet0.getLong( 1 );
            table.setRowCount( rowCount );

            //            //Capacity.
            //            resultSet0 = dataSource.getConnection().createStatement().executeQuery( "select * from " + tableName );
            //            resultSet0.

            //get all columns in one table.
            ResultSet columnSet = metaData.getColumns( null, "%", tableName, "%" );
            Column column;
            while( columnSet.next() )
            {
                //get properties of current column.
                String columnName = columnSet.getString( "COLUMN_NAME" );

                //create new cloumn.
                column = new Column( columnName, table );
                column.addProperties( "COLUMN_NAME", columnName );

                String columnType = columnSet.getString( "TYPE_NAME" );
                column.addProperties( "TYPE_NAME", columnType );

                String dataSize = String.valueOf( columnSet.getInt( "COLUMN_SIZE" ) );
                column.addProperties( "COLUMN_SIZE", dataSize );

                String digits = String.valueOf( columnSet.getInt( "DECIMAL_DIGITS" ) );
                column.addProperties( "DECIMAL_DIGITS", digits );

                String nullable = String.valueOf( columnSet.getInt( "NULLABLE" ) );
                column.addProperties( "NULLABLE", nullable );

                table.addColumn( columnName, column );
            }

            //Add FK to table
            ResultSet primaryKeySet = metaData.getPrimaryKeys( null, null, tableName );
            while( primaryKeySet.next() )
            {
                String pkName = primaryKeySet.getString( "COLUMN_NAME" );

                table.columns.get( pkName ).isFk = true;

                //TODO to be implemented.
                //                if( table.columns.get( pkName ).beReferenced == null )
                //                {
                //                    table.columns.get( pkName ).beReferenced = new LinkedList<String[]>();
                //                }
                //                

                if( table.pkColumns == null )
                {
                    table.pkColumns = new LinkedList<Schema.Column>();
                }
                table.pkColumns.add( table.columns.get( pkName ) );
            }

            //Add PK to table
            ResultSet foreignKeySet = metaData.getImportedKeys( null, null, tableName );
            while( foreignKeySet.next() )
            {
                String fkName = foreignKeySet.getString( "FKCOLUMN_NAME" );
                String pkTablenName = foreignKeySet.getString( "PKTABLE_NAME" );
                String pkColumnName = foreignKeySet.getString( "PKCOLUMN_NAME" );

                table.columns.get( fkName ).isPk = true;

                table.columns.get( fkName ).referenceTo = new String[] { pkTablenName, pkColumnName };

                if( table.fkColumns == null )
                {
                    table.fkColumns = new LinkedList<Schema.Column>();
                }
                table.fkColumns.add( table.columns.get( fkName ) );
            }

            tables.put( tableName, table );
        }
    }

    public String getUserContextName()
    {
        return userContextName;
    }

    private void metaData() throws SQLException
    {
        metaData = dataSource.getConnection().getMetaData();
    }

    public int getTableCount()
    {
        return TableAcount;
    }

    public String getName()
    {
        return name;
    }

    public Map<String, Table> getTables()
    {
        return tables;
    }

    public DataSource getDataSource()
    {
        return dataSource;
    }

    public class Table
    {
        private String hiveTableName;
        private String hdfsPath;
        private int columnCount;
        private long capacity;
        private long rowCount;

        protected String tableName;
        protected Map<String, Column> columns;
        protected LinkedList<String> columnsOrder;
        protected LinkedList<Column> pkColumns;
        protected LinkedList<Column> fkColumns;

        protected Table(String tableName)
        {
            this.tableName = tableName;
        }

        public String getName()
        {
            return tableName;
        }

        protected void addColumn( String name, Column column )
        {
            columnCount++;
            if( columns == null )
            {
                columns = new HashMap<String, Schema.Column>();
            }
            columns.put( name, column );

            if( columnsOrder == null )
            {
                columnsOrder = new LinkedList<String>();
            }
            columnsOrder.add( name );
        }

        public LinkedList<String> getColumnsOrder()
        {
            return columnsOrder;
        }

        public long getCapacity()
        {
            return capacity;
        }

        public void setCapacity( long capacity )
        {
            this.capacity = capacity;
        }

        public Schema getSchema()
        {
            return Schema.this;
        }

        public String getRowCount()
        {
            return String.valueOf( rowCount );
        }

        public void setRowCount( long rowCount )
        {
            this.rowCount = rowCount;
        }

        public String getColumnCount()
        {
            return String.valueOf( columnCount );
        }

        public String getHiveTableName()
        {
            return hiveTableName;
        }

        public void setHiveTableName( String hiveTableName )
        {
            this.hiveTableName = hiveTableName;
        }

        public String getHdfsPath()
        {
            return hdfsPath;
        }

        public void setHdfsPath( String hdfsPath )
        {
            this.hdfsPath = hdfsPath;
        }

        public Map<String, Column> getColumns()
        {
            return columns;
        }

        public LinkedList<Column> getPkColumns()
        {
            return pkColumns;
        }

        public LinkedList<Column> getFkColumns()
        {
            return fkColumns;
        }

        public boolean canJoin( Table table )
        {
            for( Column fk : table.getFkColumns() )
            {
                for( Column pk : getPkColumns() )
                {
                    if( fk.canReferenceTo( this, pk ) )
                    {
                        return true;
                    }
                }
            }
            return false;
        }

        public boolean canJoin( Table... tables )
        {
            //TODO
            return false;
        }
    }

    public class Column
    {

        public static final String COLUMN_NAME = "COLUMN_NAME";
        public static final String TYPE_NAME = "TYPE_NAME";
        public static final String COLUMN_SIZE = "COLUMN_SIZE";
        public static final String DECIMAL_DIGITS = "DECIMAL_DIGITS";
        public static final String NULLABLE = "NULLABLE";

        protected Table parentTable;
        protected String columnName;
        protected Map<String, String> properties;
        private long capacity;
        private String dataType;

        /**
         * <strong>Haven't been implemented.</strong> One column could be referenced to by several
         * other columns.
         * 
         * <pre>
         *      LinkedList{
         *                 Object[]{Table1, column1},
         *                 Object[]{Table2, column2}, 
         *                Object[]{Table2, column2}, 
         *                 ...
         *               }
         * </pre>
         */
        protected LinkedList<String[]> beReferenced;

        /**
         * One column could just reference to only another one column,
         * 
         * <pre>
         *           Object[]{Table, Column}
         * </pre>
         */
        protected String[] referenceTo;

        protected boolean isPk;
        protected boolean isFk;

        protected Column(String columnName, Table parentTable)
        {
            this.columnName = columnName;
            this.parentTable = parentTable;
        }

        protected void addReferenced( String tableName, String columnName )
        {
            if( beReferenced == null )
            {
                beReferenced = new LinkedList<String[]>();
            }

            beReferenced.add( new String[] { tableName, columnName } );
        }

        protected void setRefenceTo( String tableName, String columnName )
        {
            referenceTo = new String[] { tableName, columnName };
        }

        protected void addReferenced( Table table, Column column )
        {
            addReferenced( table.getName(), column.getName() );
        }

        protected void setRefenceTo( Table table, Column column )
        {
            setRefenceTo( table.getName(), column.getName() );
        }

        public String getName()
        {
            return columnName;
        }

        public long getCapacity()
        {
            return capacity;
        }

        public void setCapacity( long capacity )
        {
            this.capacity = capacity;
        }

        protected void addProperties( String name, String property )
        {
            if( properties == null )
            {
                properties = new HashMap<String, String>();
            }
            properties.put( name, property );
        }

        /**
         * <strong>Warn!!! Don't use this method, the essential implementation in
         * <code>Column</code> haven't been implemented.</strong> Evaluate that whether the column
         * of given table could reference to current column or not.
         * */
        public boolean canBeReferenced( Table table, Column column )
        {

            for( String[] o : beReferenced )
            {
                //If the table name and the column name all are the same, then return true.
                if( o[0].equalsIgnoreCase( table.getName() ) && o[1].equalsIgnoreCase( column.getName() ) )
                {
                    return true;
                }
            }

            return false;
        }

        /**
         * Evaluate that whether the column of given table could be referenced by current column or
         * not.
         * */
        public boolean canReferenceTo( Table table, Column column )
        {
            return referenceTo[0].equalsIgnoreCase( table.getName() )
                && referenceTo[1].equalsIgnoreCase( column.getName() );
        }

        public Table getParentTable()
        {
            return parentTable;
        }

        public String getColumnName()
        {
            return columnName;
        }

        /**
         * Use static final fields to get properties.
         * */
        public String getProperty( String Property )
        {
            return properties.get( Property );
        }

        public boolean isFK()
        {
            return isFk;
        }

        public boolean isPK()
        {
            return isPk;
        }
    }
}
